package com.huawei.bigdata.kafka.temp;

import com.huawei.bigdata.kafka.example.security.LoginUtil;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

@Slf4j
@Service
public class TestProducer {

    private KafkaProducer<String, String> producer;
    @Value("${conf.kafka.test.user-keytab-file}")
    private String userKeytabFile;
    @Value("${conf.kafka.test.user-principal}")
    private String userPrincipal;
    @Value("${conf.kafka.test.security}")
    private Boolean security;
    @Resource(name = "initProperties")
    private Properties props;

    @PostConstruct
    public void init(){
        log.info("-=+props: {}", props);
        if(this.producer == null){
            if(security){
                log.info("auth");
                if (LoginUtil.isSecurityModel()) {
                    try {
                        log.info("userPrincipal：{}, userKeytabFile:{}", userPrincipal, userKeytabFile);

                        // !!注意，安全认证时，需要用户手动修改为自己申请的机机账号
                        LoginUtil.securityPrepare(userPrincipal, userKeytabFile);
                    } catch (IOException e) {
                        log.error("Security prepare failure.");
                        log.error("The IOException occured.", e);
                        return;
                    }
                    log.info("Security prepare success.");
                }
            }
            try {
                this.producer = new KafkaProducer<>(props);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }


    public void send(String topic, String key, String value, int isAsync){

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        long startTime = System.currentTimeMillis();
        if (isAsync==1) {
            // 异步发送
            producer.send(record, (metadata, exception)-> {
                if(exception != null){
                    log.info("-=+send-msg-fail\nThe message was sent to tipic:{} partition: {}, offset: {} in {} ms"
                            , topic, metadata.partition(), metadata.offset(), System.currentTimeMillis() - startTime, exception);
                }else {
                    log.info("The message was sent to tipic:{} partition: {}, offset: {} in {} ms", topic, metadata.partition(), metadata.offset(), System.currentTimeMillis() - startTime);
                }

            });
        } else {
            try {
                // 同步发送
                producer.send(record).get();
            } catch (InterruptedException ie) {
                log.info("The InterruptedException occured : {}.", ie);
            } catch (ExecutionException ee) {
                log.info("The ExecutionException occured : {}.", ee);
            }
        }
    }
}
